package com.example.kafka.service;

import com.example.kafka.config.KafkaConfig;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.stereotype.Service;

import java.util.Properties;

/**
 * 低基数分区键 Service
 */
@Service
public class PartitionKeyLeakService {
    public void send1(String status, String msg) {
        // 使用状态码作为分区键，基数极低，极易数据倾斜
        KafkaProducer<String, String> producer = new KafkaProducer<>(KafkaConfig.getProducerProps());
        producer.send(new ProducerRecord<>("topic_partition", status, msg));
        // 未关闭 producer
    }

    public void send2(String time, String msg) {
        // 使用时间戳作为分区键，短时间内数据全部写入同一分区
        KafkaProducer<String, String> producer = new KafkaProducer<>(KafkaConfig.getProducerProps());
        producer.send(new ProducerRecord<>("topic_partition", time, msg));
        // 未关闭 producer
    }

    public void send3(String msg) {
        // 分区键为 null，采用轮询策略，可能与业务预期不符
        KafkaProducer<String, String> producer = new KafkaProducer<>(KafkaConfig.getProducerProps());
        producer.send(new ProducerRecord<>("topic_partition", null, msg));
        // 未关闭 producer
    }

    public void send4(String productId, String msg) {
        // 热门产品ID作为分区键，热点值导致数据倾斜
        KafkaProducer<String, String> producer = new KafkaProducer<>(KafkaConfig.getProducerProps());
        producer.send(new ProducerRecord<>("topic_partition", productId, msg));
        // 未关闭 producer
    }

    public void send5(String key, String msg) {
        // 自定义分区器实现不均衡（如所有key都分到0号分区）
        Properties props = KafkaConfig.getProducerProps();
        props.put("partitioner.class", "com.example.kafka.service.BadPartitioner");
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        producer.send(new ProducerRecord<>("topic_partition", key, msg));
        // 未关闭 producer
    }
}
